package com.learn.lb.mapred;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.stream.Stream;

/**
 * @author laibo
 * @since 2019/7/3 15:48
 * ===================================
 * <p>统计需求：
 * <p>
 * 1、统计每门课程的参考人数和课程平均分
 * 2、统计每门课程参考学生的平均分，并且按课程存入不同的结果文件，要求一门课程一个结果文件，并且按平均分从高到低排序，分数保留一位小数。
 * 3、求出每门课程参考学生成绩最高的学生的信息：课程，姓名和平均分。
 * </p>
 * 数据格式存放在resources/score,
 * 第一个字段表示学科，第二个字段表示学生姓名，后面表示每次考试的成绩，个数不固定
 * ===================================
 */
public class ScoreGroupTask {

    private static Logger logger = LoggerFactory.getLogger(ScoreGroupTask.class);

    /**
     * 统计每门课程的参考人数和课程平均分
     */
    public static class AvgMap extends Mapper<LongWritable, Text, Text, DoubleWritable> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //value表示分片后一行的数据
            String[] source = value.toString().trim().split(",");
            //因为要以学科分组，所以输出的key应该是学科，也就是第一个参数
            Text outKey = new Text(source[0]);
            //输出的value约定格式：
            if (source.length < 3) {
                //如果这个学生没有考试过，则直接返回0分
                context.write(outKey, new DoubleWritable(0D));
                return;
            }
            //计算该学科总分
            double sumScore = Stream.of(source).filter(StringUtils::isNumeric).mapToDouble(Double::valueOf).sum();
            logger.info("姓名为：{} 的课程{}的考试总分为：{}", source[0], source[1], sumScore);
            context.write(outKey, new DoubleWritable(sumScore));
        }
    }

    public static class AvgReducer extends Reducer<Text, DoubleWritable, Text, Text> {
        @Override
        protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
            int countStudent = 0;
            double sumScore = 0D;
            for (DoubleWritable value : values) {
                countStudent++;
                sumScore += value.get();
            }
            //输出  学科  人数  平均分
            context.write(key, new Text("\t" + countStudent + "\t" + Math.round(sumScore / countStudent)));
        }
    }

    public static void main(String[] args) throws Exception {
        logger.info("开始运行mapreducer任务.....");
        Configuration hdfsConfiguration = new Configuration();
        Path dstPath = new Path("/score_in/input_data");
        Path outPath = new Path("/score_out");
        GenerateMapReducerUtils.buildJobBefore(hdfsConfiguration, "score.txt", dstPath, outPath);
        logger.info("开始进行MapReducer计算....");
        Job job = Job.getInstance(hdfsConfiguration, "Score Avg");
        job.setJarByClass(ScoreGroupTask.class);
        job.setMapperClass(AvgMap.class);
        job.setReducerClass(AvgReducer.class);
        //设置输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        //设置输出目录
        FileInputFormat.addInputPath(job, dstPath);
        FileOutputFormat.setOutputPath(job, outPath);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }


}
